package cn.zl.orchestra.task;

import cn.zl.orchestra.common.ApiMeta;
import cn.zl.orchestra.common.ContextMeta;
import cn.zl.orchestra.common.OrchestraDag;
import cn.zl.orchestra.common.ParamMeta;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.JsonPath;
import com.mysql.cj.jdbc.MysqlDataSource;
import groovy.lang.GroovyClassLoader;
import groovy.lang.GroovyObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 先将结果收集成list形式，然后进行流式处理
 *
 * @author zhong lin  2022/8/8
 */
@Slf4j
public class TaskOrchestraDemo {

    private static final JSONObject context = new JSONObject();

    private static final Map<String, Object> resultMap = new HashMap<>(8);

    public Map<String, Object> scheduleApi(ContextMeta meta) {

        // 将dag图转为流形式的
        OrchestraDag orchestraDag = new OrchestraDag(meta.getEdgeList());

        List<ApiMeta> baseDataQueryMeta = meta.getBaseDataQueryMeta();
        Map<Long, ApiMeta> apiIdMap = baseDataQueryMeta.stream().collect(Collectors.toMap(ApiMeta::getApiId, e -> e));

        // 根据流结构迭代处理各个节点
        while (orchestraDag.haveNext()) {
            Long currentApiId = orchestraDag.popApiId();

            ApiMeta apiMeta = apiIdMap.get(currentApiId);

            String apiType = apiMeta.getApiType();

            switch (apiType) {
                case "Customize_SQL_API":
                case "Register_API":
                    sqlQuery(apiMeta);
                    break;
                case "function":
                    functionHandle(apiMeta);
                    break;
                case "start":
                    startNodeHandle(apiMeta, meta.getRequestParams());
                    break;
                case "end":
                    endNodeHandle(apiMeta);
                    break;
                default:
                    throw new RuntimeException("不支持的api类型");
            }

            // todo 此处应该增加一个判断，根据当前的 api ID 判断其父节点是否还有子节点没有执行，
            //  如果父节点全部的子节点都执行完毕了。则删除上下文中，当父节点的信息，避免数据太大
            // start 节点例外
        }

        return resultMap;
    }

    /**
     * end 节点 -- 根据end  节点定义的参数，获取服务编排API的响应参数
     *
     * @param apiMeta
     */
    public static void endNodeHandle(ApiMeta apiMeta) {
        List<ParamMeta> defineParameterMetaList = apiMeta.getDefineParameterMetaList();

        DocumentContext contextParse = JsonPath.parse(context.toJSONString());
        defineParameterMetaList.forEach(paramMeta -> {
            String paramMappingJsonPath = paramMeta.getParamMapping();
            Object paramValue = paramMeta.getParamValue();
            if (StringUtils.isNotBlank(paramMappingJsonPath)) {
                paramValue = contextParse.read(paramMappingJsonPath);
            }
            resultMap.put(paramMeta.getParamName(), paramValue);
        });
    }


    /**
     * start node 结合请求参数将start 节点的参数放到上下文信息中
     *
     * @param apiMeta
     * @param requestParams
     */
    public static void startNodeHandle(ApiMeta apiMeta, List<ParamMeta> requestParams) {
        List<ParamMeta> defineParameterMetaList = apiMeta.getDefineParameterMetaList();
        if (CollectionUtils.isEmpty(defineParameterMetaList)) {
            return;
        }

        // 处理开始节点参数信息
        Map<String, Object> kvMap = requestParams.stream().collect(Collectors.toMap(ParamMeta::getParamName,
                ParamMeta::getParamValue));
        Map<String, Object> startParam = new HashMap<>();
        defineParameterMetaList.forEach(paramMeta -> {
            String paramName = paramMeta.getParamName();
            startParam.put(paramName, kvMap.get(paramName));
        });
        // 将请求参数放置到上下文中 -- 跨节点使用
        context.put("start", startParam);
    }


    /**
     * udf 函数 -- 调用函数方法执行
     *
     * 输入一条数据产生一条数据
     *
     * 后续需要针对不同的函数进行判断，根据函数的类型与分类进行处理
     */
    public static void functionHandle(ApiMeta apiMeta) {
        // 获取函数的入参
        Map<String, Object> paramMeta = parseFunctionInputParam(apiMeta);
        List<Object> param = new ArrayList<>();
        if (MapUtils.isNotEmpty(paramMeta)) {
            param = new ArrayList<>(paramMeta.values());
        }

        String groovy = apiMeta.getFunctionContent();
        GroovyClassLoader classLoader = new GroovyClassLoader();
        Class groovyClass = classLoader.parseClass(groovy);
        GroovyObject groovyObject = null;
        try {
            groovyObject = (GroovyObject) groovyClass.newInstance();
            final String result = (String) groovyObject.invokeMethod("demo", param.toArray());
            final JSONObject resultObj = JSON.parseObject(result);
            // todo 函数输出的是一个json形式。如果用户定义了函数的输出，则从函数输出的json中取出相应的key，放到结果集中。
            // 并非像现在这样，将整个输出结果存放到上下文中

            context.put(apiMeta.getApiName(), resultObj);
        } catch (Exception e) {
            log.info("执行函数失败");
        }
    }



    public static Map<String, Object> parseFunctionInputParam(ApiMeta apiMeta) {
        List<ParamMeta> defineParameterMetaList = apiMeta.getDefineParameterMetaList();
        if (CollectionUtils.isEmpty(defineParameterMetaList)) {
            return Collections.emptyMap();
        }

        Map<String, Object> paramMap = new HashMap<>(8);

        DocumentContext parseContext = JsonPath.parse(context);

        defineParameterMetaList.stream()
                .filter(paramMeta -> Objects.equals(paramMeta.getParamPosition(), "input"))
                .forEach(paramMeta -> {
                    String paramMappingJsonPath = paramMeta.getParamMapping();
                    Object paramValue = paramMeta.getParamValue();
                    if (StringUtils.isNotBlank(paramMappingJsonPath)) {
                        paramValue = parseContext.read(paramMappingJsonPath);
                    }
                    paramMap.put(paramMeta.getParamName(), paramValue);
                });

        return paramMap;
    }


    public static Map<String, Object> parseParam(ApiMeta apiMeta) {
        List<ParamMeta> defineParameterMetaList = apiMeta.getDefineParameterMetaList();
        if (CollectionUtils.isEmpty(defineParameterMetaList)) {
            return Collections.emptyMap();
        }

        Map<String, Object> paramMap = new HashMap<>(8);

        DocumentContext parseContext = JsonPath.parse(context);

        defineParameterMetaList.forEach(paramMeta -> {
            String paramMappingJsonPath = paramMeta.getParamMapping();
            Object paramValue = paramMeta.getParamValue();
            if (StringUtils.isNotBlank(paramMappingJsonPath)) {
                paramValue = parseContext.read(paramMappingJsonPath);
            }

            paramMap.put(paramMeta.getParamName(), paramValue);
        });

        return paramMap;
    }


    /**
     * API节点 -- 调用API服务执行
     *
     * @param apiMeta
     * @return
     */
    public static void sqlQuery(ApiMeta apiMeta) {

        String sql = apiMeta.getSqlContent();

        // 此处仅仅作参数解析的展示，参数的拼接、处理等，已经有现成的代码。此处不再展示
        Map<String, Object> paramMeta = parseParam(apiMeta);

        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;

        try {
            connection = buildDatasource(apiMeta).getConnection();
            preparedStatement = connection.prepareStatement(sql);
            resultSet = preparedStatement.executeQuery();

            final List<Map<String, Object>> maps = buildResult(resultSet);

            context.put(apiMeta.getApiName(), maps);
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                if (Objects.nonNull(resultSet)) {
                    resultSet.close();
                }

                if (Objects.nonNull(preparedStatement)) {
                    preparedStatement.close();
                }

                if (Objects.nonNull(connection)) {
                    connection.close();
                }

            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    public static DataSource buildDatasource(ApiMeta apiMeta) {
        MysqlDataSource mysqlDataSource = new MysqlDataSource();
        mysqlDataSource.setUrl(apiMeta.getConnectUrl());
        mysqlDataSource.setPassword(apiMeta.getPassword());
        mysqlDataSource.setUser(apiMeta.getUsername());

        return mysqlDataSource;
    }


    private static List<Map<String, Object>> buildResult(ResultSet resultSet) throws SQLException {
        Set<String> resultColumns = buildResultColumn(resultSet.getMetaData());
        List<Map<String, Object>> resultValues = new LinkedList<>();

        int rowCount = 0;
        while (!resultSet.isClosed() && resultSet.next() && rowCount < 10) {
            Map<String, Object> rowValues = new LinkedHashMap<>(resultColumns.size());
            resultValues.add(rowValues);
            rowCount++;
            for (String columnName : resultColumns) {
                Object columnValue = resultSet.getObject(columnName);

                rowValues.put(columnName, columnValue);
            }
        }
        return resultValues;
    }

    protected static Set<String> buildResultColumn(ResultSetMetaData rsm) throws SQLException {
        int fieldCnt = rsm.getColumnCount();
        Set<String> resultColumns = new LinkedHashSet<>();
        for (int i = 1; i <= fieldCnt; i++) {
            String columnName = rsm.getColumnLabel(i);
            resultColumns.add(columnName);
        }
        return resultColumns;
    }

}
